package com.alibaba.citrus.cr.mn.order.center.facade.function.impl;

import com.alibaba.citrus.cr.b2b.trade.order.repository.OrderBackUpRepository;
import com.alibaba.citrus.cr.b2b.trade.order.repository.OrderQueryServiceRepository;
import com.alibaba.citrus.cr.common.enums.ordercenter.InnerMessageEventEnum;
import com.alibaba.citrus.cr.mn.order.center.facade.function.common.SendMsgService;
import com.alibaba.citrus.cr.mn.order.center.facade.kafka.KafkaUtils;
import com.alibaba.citrus.cr.mn.order.center.forecastorder.service.ForecastOrderBizQueryService;
import com.alibaba.cz.base.tool.log.Log;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.lattice2.epoch.db.service.CustomSequenceAccessor;
import com.alibaba.lattice2.epoch.middleware.mq.runtime.EpochMessageService;
import com.epoch.app.crb2btradecenter.domain.order.dto.QueryByOrderBizIdsRequest;
import com.epoch.app.crb2btradecenter.messagetask.model.MessageTask;
import com.epoch.app.crb2btradecenter.messagetask.service.MessageTaskService;
import com.epoch.app.crb2btradecenter.model.dto.TradeMessageSDO;
import com.epoch.app.crb2btradecenter.model.enums.MessageTaskTypeEnum;
import com.epoch.app.crb2btradecenter.sdo.OrderLineSDO;
import com.epoch.app.crb2btradecenter.sdo.OrderSDO;
import com.epoch.app.crforecastsales.model.dto.FailOrderRequest;
import com.epoch.app.crforecastsales.sdo.ForecastOrderSDO;
import com.epoch.app.mnordercenter.contants.MnOrderCenterConstants;
import com.epoch.app.mnordercenter.model.dto.CommonAsyncMessageResponse;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author zhangxiaorui
 * @ClassName SalesOrderAtomCreateSendMQServiceImpl
 * @description: TODO
 * @date 2023/04/21 19:27
 * @version: 1.0
 */
@Service
public class SendMsgServiceImpl implements SendMsgService {
    private static final Log log = Log.getLogger(SendMsgServiceImpl.class);

    @Resource
    private CustomSequenceAccessor customSequenceAccessor;
    @Resource
    private EpochMessageService messageService;
    @Resource
    private MessageTaskService messageTaskService;
    @Resource
    private OrderQueryServiceRepository orderQueryServiceRepository;
    @Resource
    private OrderBackUpRepository ordersBackUpRepository;
    @Resource
    private ForecastOrderBizQueryService forecastOrderBizQueryService;

    // topic
    private static final String SALES_MQ_TOPIC = "CR_SALEORDER_V2";
    private static final String FORECAST_MQ_TOPIC = "CR_FORECAST_V2";

    // group
    private static final String SALE_ORDER_MESSAGE = "sale_order_message"; // GID_SALEORDER
    private static final String FORECAST_ORDER_MESSAGE = "forecast_send_message";

    // tag
    private static final String SALES_BATCH_CREATED_TAG = "SaleBatchCreated";
    private static final String SALES_CREATED_TAG = "SaleCreated";
    private static final String SALES_REFUSED_TAG = "SaleRefused";
    private static final String SALES_ORDER_PUSH_MN_TAG = "SaleOrderPushMn";
    private static final String SALES_ORDER_MODIFY_TAG = "SalesOrderModify";
    private static final String SALES_ORDER_EVENT_TAG = "SalesOrderEvent";
    private static final String FORECAST_ORDER_CLOSED_TAG = "ForecastOrderClosed";
    private static final String EX_SALE_ORDER_IN_COMPLETE_TAG = "EXSaleOrderInComplete";
    private static final String FORECAST_ORDER_CHANGE_TAG = "ForecastOrderChange";
    private static final String SALE_CREATE_FAILED_TAG = "SALE_CREATE_FAILED";
    private static final String SALE_ORDER_CONFIRM_QUANTITY_TAG = "SalesOrderConfirmQuantity";
    private static final String SALES_ORDER_REFUSE_TAG = "SalesOrderRefuse";
    private static Map<String, String> getMQProperties() {
        Map<String, String> map = Maps.newHashMap();
        map.put("_CZ_KEY", "mengniu");
        return map;
    }

    /**
     * 发送消息推送订单给SAP
     */
    @Override
    public void startTaskAndSendSalesOrderMessage(Map<String, List<String>> confirmOrderMap) {
        log.info("SendMsgServiceImpl#startTaskAndSendSalesOrderMessage confirmOrderMap={}", JSON.toJSONString(confirmOrderMap));
        if (MapUtils.isEmpty(confirmOrderMap)) {
            log.info("SendMsgServiceImpl#startTaskAndSendSalesOrderMessage confirmOrderMap is empty");
            return;
        }
        String sequenceNo = customSequenceAccessor.getNextStringSequence("cr_b2b_trade_center", "batchCreateOrderSequence");
        //生成销售单对象子单总条数
        Integer orderDetailSize = confirmOrderMap.values().stream().map(List::size).reduce(0, Integer::sum);
        MessageTask messageTask = new MessageTask();
        messageTask.setBatchNo(sequenceNo);
        messageTask.setType(MessageTaskTypeEnum.ORDER_BATCH_SUBMIT.getValue());
        messageTask.setExtraContent(JSON.toJSONString(confirmOrderMap));
        messageTask.setExtraCount(orderDetailSize);
        messageTask.setGmt_create(new Date());
        messageTask.setGmt_modified(new Date());
        messageTask.setStatus(1);
        log.info("SendMsgServiceImpl#startTaskAndSendSalesOrderMessage messageTask={}", JSONObject.toJSONString(messageTask));
        String taskId = messageTaskService.create(messageTask).getId();

        Boolean batchSendResp = messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALES_BATCH_CREATED_TAG, sequenceNo, sequenceNo, getMQProperties());

        log.info("SendMsgServiceImpl#startTaskAndSendSalesOrderMessage taskId={}, sequenceNo={}, batchSendResp={}", taskId, sequenceNo, batchSendResp);
    }

    /**
     * 发消息预占库存
     */
    @Override
    public void sendMsgToInventoryOccupy(List<OrderSDO> orderSubmitList) {
        for (OrderSDO x : orderSubmitList) {
            TradeMessageSDO tradeMessageSDO = new TradeMessageSDO();
            tradeMessageSDO.setOrderSDO(x);

            String message = JSON.toJSONString(tradeMessageSDO);
            if (StringUtils.isBlank(message)) {
                continue;
            }
            Boolean sendResp = messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALES_CREATED_TAG, x.getOrderBizId(), message, getMQProperties());
            log.info("SendMsgServiceImpl#sendMsgToInventoryOccupy sendResp={}, message={}", sendResp, message);
        }
    }

    /**
     * 发消息释放库存
     */
    @Override
    public void sendMsgToInventoryRelease(List<OrderSDO> orderList) {
        for (OrderSDO orderSDO : Optional.ofNullable(orderList).orElse(Lists.newArrayList())) {
            List<String> orderLineIds = orderSDO.getOrderLineSDOS().stream().map(OrderLineSDO::getId).collect(Collectors.toList());
            TradeMessageSDO tradeMessageSDO = new TradeMessageSDO();
            tradeMessageSDO.setOrderSDO(orderSDO);
            tradeMessageSDO.setAffectLineIds(orderLineIds);

            String message = JSON.toJSONString(tradeMessageSDO);
            if (StringUtils.isBlank(message)) {
                continue;
            }
            Boolean sendResp = messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALES_REFUSED_TAG,
                    orderSDO.getOrderBizId(), JSON.toJSONString(tradeMessageSDO), getMQProperties());
            log.info("SendMsgServiceImpl#sendMsgToInventoryRelease bizOrderId={} orderLineIds={} sendResp={}",
                    orderSDO.getOrderBizId(), JSON.toJSONString(orderLineIds), sendResp);
        }
    }

    @Override
    public void sendMsgToSAP(String msg) {
        log.info("AbstractOrderAtomUpdateService#sendMsgToSAP msg={}", JSON.toJSONString(msg));
        messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALES_ORDER_MODIFY_TAG,
                UUID.randomUUID().toString(), msg, getMQProperties());
    }

    @Override
    public void sendMsgToSendKafka(CommonAsyncMessageResponse message) {
        if (Objects.isNull(message) || CollectionUtils.isEmpty(message.getData())) {
            return;
        }

        Boolean sendResult = messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALES_ORDER_PUSH_MN_TAG,
                UUID.randomUUID().toString(), JSON.toJSONString(message), getMQProperties());

        log.info("SendMsgServiceImpl#sendMsgToSendKafka message={}, sendResult={}",
                JSON.toJSONString(message), JSON.toJSONString(sendResult));
    }

    /**
     * 发消息到kafka
     */
    @Override
    public void sendMsgToKafka(String topic, String message) {
        log.info("SendMsgServiceImpl#sendMsgToKafka message: {}", message);
        if (StringUtils.isBlank(topic)) {
            topic = MnOrderCenterConstants.MN_KAFKA_TOPIC();
        }
        KafkaUtils.send(topic, message);
    }

    /**
     * 发送订单变更消息
     */
    @Override
    public void sendSalesOrderChangeMsg(List<String> salesOrderNos, InnerMessageEventEnum event) {
        log.info("SendMsgServiceImpl#sendSalesOrderChangeMsg salesOrderNos={} event={}", salesOrderNos, JSON.toJSONString(event));
        if (CollectionUtils.isEmpty(salesOrderNos)) {
            return;
        }

        if (InnerMessageEventEnum.SALES_ORDER_DELETED.equals(event)) {
            // todo@dongju 订单删除时需要查备份表
//            com.epoch.app.crb2btradecenter.orderbackup.dto.QueryByOrderBizIdsRequest request = new com.epoch.app.crb2btradecenter.orderbackup.dto.QueryByOrderBizIdsRequest();
//            request.setOrderBizIds(salesOrderNos);
//            request.setLimit(salesOrderNos.size());
//            Result<List<OrderBackUp>> orderBackUpResult = ordersBackUpRepository.queryByOrderBizIds(request);
//            if (Objects.isNull(orderBackUpResult) || CollectionUtils.isEmpty(orderBackUpResult.getResult())) {
//                return;
//            }
            return;
        }

        QueryByOrderBizIdsRequest request = new QueryByOrderBizIdsRequest();
        request.setOrderBizIds(salesOrderNos);
        List<OrderSDO> orderSDOS = orderQueryServiceRepository.queryByOrderBizIds(request, true);
        if (CollectionUtils.isEmpty(orderSDOS)) {
            log.info("SendMsgServiceImpl#sendSalesOrderChangeMsg order is empty");
            return;
        }

        CommonAsyncMessageResponse message = new CommonAsyncMessageResponse();
        message.setEventCode(event.getValue());
        message.setEventName(event.getLabel());
        message.setData(orderSDOS);
        String messageString = JSON.toJSONString(message);
        String messageKey = UUID.randomUUID().toString();

        log.info("SendMsgServiceImpl#sendSalesOrderChangeMsg key={} message={}", messageKey, messageString);
        messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALES_ORDER_EVENT_TAG,
                messageKey, messageString, getMQProperties());
    }

    @Override
    public void sendSalesOrderChangeMsg(Map<String, List<String>> salesOrderNosMap, InnerMessageEventEnum event) {
        log.info("SendMsgServiceImpl#sendSalesOrderChangeMsg salesOrderNosMap={} event={}", salesOrderNosMap, JSON.toJSONString(event));
        if (MapUtils.isEmpty(salesOrderNosMap)) {
            return;
        }
        if (InnerMessageEventEnum.SALES_ORDER_DELETED.equals(event)) {
            return;
        }
        QueryByOrderBizIdsRequest request = new QueryByOrderBizIdsRequest();
        request.setOrderBizIds(Arrays.asList(salesOrderNosMap.keySet().toArray()));
        List<OrderSDO> orderSDOS = orderQueryServiceRepository.queryByOrderBizIds(request, true);
        if (CollectionUtils.isEmpty(orderSDOS)) {
            log.info("SendMsgServiceImpl#sendSalesOrderChangeMsg order is empty");
            return;
        }
        if (InnerMessageEventEnum.SALES_ORDER_REFUSED.equals(event) || InnerMessageEventEnum.SALES_ORDER_CANCELED.equals(event)) {
            for (OrderSDO orderSDO : orderSDOS) {
                List<OrderLineSDO> orderLine = new ArrayList();
                List<OrderLineSDO> orderLineSDOS = orderSDO.getOrderLineSDOS();
                List<String> lineNumList = salesOrderNosMap.get(orderSDO.getOrderBizId());
                for (OrderLineSDO orderLineSDO : orderLineSDOS) {
                    if (lineNumList.contains(orderLineSDO.getSerialNumber())) {
                        orderLine.add(orderLineSDO);
                    }
                }
                orderSDO.setOrderLineSDOS(orderLine);
            }
        }
        CommonAsyncMessageResponse message = new CommonAsyncMessageResponse();
        message.setEventCode(event.getValue());
        message.setEventName(event.getLabel());
        message.setData(orderSDOS);
        String messageString = JSON.toJSONString(message);
        String messageKey = UUID.randomUUID().toString();

        log.info("SendMsgServiceImpl#sendSalesOrderChangeMsg key={} message={}", messageKey, messageString);
        messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALES_ORDER_EVENT_TAG,
                messageKey, messageString, getMQProperties());
    }

    /**
     * 发送预报订单变更消息
     */
    @Override
    public void sendForecastOrderChangeMsg(List<String> forecastOrderNoList, InnerMessageEventEnum event) {
        log.info("SendMsgServiceImpl#sendForecastOrderChangeMsg salesOrderNos={} event={}", forecastOrderNoList, JSON.toJSONString(event));
        if (CollectionUtils.isEmpty(forecastOrderNoList)) {
            return;
        }
        List<ForecastOrderSDO> forecastOrderSDOList = forecastOrderBizQueryService.queryByOrderBizNo(forecastOrderNoList);
        if (CollectionUtils.isEmpty(forecastOrderSDOList)) {
            log.info("SendMsgServiceImpl#sendForecastOrderChangeMsg order is empty");
            return;
        }
        CommonAsyncMessageResponse message = new CommonAsyncMessageResponse();
        message.setEventCode(event.getValue());
        message.setEventName(event.getLabel());
        message.setData(forecastOrderSDOList);
        String messageString = JSON.toJSONString(message);
        String messageKey = UUID.randomUUID().toString();
        log.info("SendMsgServiceImpl#sendOrderChangeMsg key={} message={}", messageKey, messageString);
        messageService.sendMessage(FORECAST_ORDER_MESSAGE, FORECAST_MQ_TOPIC, FORECAST_ORDER_CHANGE_TAG, messageKey, messageString, getMQProperties());
    }

    @Override
    public void sendForecastClosedMsgToSAP(String message) {
        try {
            messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, FORECAST_ORDER_CLOSED_TAG,
                    UUID.randomUUID().toString(), message, getMQProperties());
        } catch (Exception e) {
            log.error("预报订单批量关闭发送消息失败", e);
        }
    }

    @Override
    public void sendExSaleOrderInComplete(OrderSDO orderSDO) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("orderBizId", orderSDO.getOrderBizId());
        String messageKey = UUID.randomUUID().toString();
        log.info("SendMsgServiceImpl#sendExSaleOrderInComplete key={} message={}", messageKey, jsonObject.toJSONString());
        messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, EX_SALE_ORDER_IN_COMPLETE_TAG,
                messageKey, jsonObject.toJSONString(), getMQProperties());
    }

    @Override
    public void sendSalesOrderRefuseMsg(String message) {
        String messageKey = UUID.randomUUID().toString();
        log.info("SendMsgServiceImpl#sendOrderRefuseMsg key={} message={}", messageKey, message);
        messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALES_ORDER_REFUSE_TAG,
                messageKey, message, getMQProperties());
    }

    @Override
    public void sendActivityFailMsg(String sequenceNo, List<Long> forecastLineIds) {
        log.info("SendMsgServiceImpl#sendAssembleActivityFailMsg sequenceNo={} forecastLineIds={}", sequenceNo, forecastLineIds);
        if (StringUtils.isBlank(sequenceNo) || CollectionUtils.isEmpty(forecastLineIds)) {
            return;
        }
        FailOrderRequest failOrderRequest = new FailOrderRequest();
        failOrderRequest.setSequenceNo(sequenceNo);
        failOrderRequest.setFailOrderIds(forecastLineIds.stream().distinct().collect(Collectors.toList()));
        messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALE_CREATE_FAILED_TAG, JSON.toJSONString(failOrderRequest));
    }

    @Override
    public void sendConfirmQuantityMsg(String message) {
        log.info("SendMsgServiceImpl sendConfirmQuantityMsg message={}",message);
        messageService.sendMessage(SALE_ORDER_MESSAGE, SALES_MQ_TOPIC, SALE_ORDER_CONFIRM_QUANTITY_TAG, message);
    }
}
